package com.xzx.flume;

import lombok.extern.slf4j.Slf4j;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;

import java.util.Random;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2025/5/11 22:37
 */
@Slf4j
public class MySource extends AbstractSource implements Configurable, PollableSource {
    private Long sleep = 1000L;
    /**
     * 获取配用的
     *
     * @param context
     */
    @Override
    public void configure(Context context) {
        sleep = context.getLong("sleep", 1000 * 10L);
        log.info("source的参数：sleep:{}", sleep);
    }

    /**
     * 生产数据
     */
    @Override
    public Status process() throws EventDeliveryException {
        Status status = null;

        try {
            // This try clause includes whatever Channel/Event operations you want to do

            // Receive new data
            Event e = getSomeData();

            // Store the Event into this Source's associated Channel(s)
            getChannelProcessor().processEvent(e);

            status = Status.READY;
        } catch (Throwable t) {
            // 如果发生异常，channel事务会回滚，然后这里source可以按需回滚

            status = Status.BACKOFF;

            // re-throw all Errors
            if (t instanceof Error) {
                throw (Error) t;
            }
        }
        return status;
    }

    private Event getSomeData() throws InterruptedException {
        Thread.sleep(sleep);
        SimpleEvent simpleEvent = new SimpleEvent();
        simpleEvent.setBody(("我的自定义source:" + new Random().nextInt(100)).getBytes());
        simpleEvent.getHeaders().put("myHeader", "myHeaderValue");
        return simpleEvent;
    }

    /**
     * 如果连续回滚了。每次回滚后中间休眠的时间会增长，每次增长的步长
     */
    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    /**
     * 如果连续回滚，每次回滚之间休眠时间会增加，但是不能超过这个方法设置的最大时间
     */
    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }


}
